package start.demo01.datatype.json.entity;

import com.alibaba.fastjson.JSONObject;
import lombok.Data;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import start.demo01.datatype.datasource.DataSourcePool;
import start.demo01.datatype.datasource.ESPool;
import start.demo01.datatype.datasource.KafkaPool;
import start.demo01.datatype.datasource.SyslogPool;
import start.demo01.datatype.json.pool.CacheMap;

import javax.sql.DataSource;
import java.io.IOException;
import java.sql.SQLException;

@Data
public class OutputConfig {

    private Boolean mysql=false;
    private Boolean oracle=false;
    private Boolean es=false;
    private Boolean kafka=false;
    private Boolean syslog=false;

    private DataSourcePool dataSourcePool;
    private ESPool esPool;
    private KafkaPool kafkaPool;

    public OutputConfig() {
        dataSourcePool=new DataSourcePool();
        esPool=new ESPool();
        kafkaPool=new KafkaPool();
    }

    public void exec(ConfigInfo configInfo,CacheMap cache) throws SQLException, IOException {
        if(es){
            System.out.println("ES插入");
            RestHighLevelClient client = esPool.getDataSource("es");
            IndexRequest request=new IndexRequest(configInfo.getEsInfo().getIndex());
            request.source(((JSONObject)cache.get("es")).toJSONString(), XContentType.JSON);
            client.index(request, RequestOptions.DEFAULT);
        }
        if(mysql){
            System.out.println("mysql插入");
            String sql = cache.get("sql").toString();
            sql=sql.replace("?",configInfo.getTableInfo().getName());
            DataSource dataSource = dataSourcePool.getDataSource("key1");
            dataSource.getConnection()
                    .createStatement()
                    .executeUpdate(sql);
        }
        if(kafka){
            System.out.println("Kafka插入");
            ConfigInfo.KafkaInfo kafkaInfo = configInfo.getKafkaInfo();
            kafkaPool.send(kafkaPool.producer(kafkaInfo.getServers()),
                    kafkaInfo.getTopic(),
                    ((JSONObject)cache.get("kafka")).toJSONString());
        }
        if(syslog){
            System.out.println("syslog插入");
            ConfigInfo.SyslogInfo syslogInfo = configInfo.getSyslogInfo();
            SyslogPool.sendLog(syslogInfo.getProtocol(),
                    syslogInfo.getIp(), syslogInfo.getPort(),
                    ((JSONObject)cache.get("syslog")).toJSONString(), syslogInfo.getLevel());
        }
    }
}
